package mr

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"sort"
	"strings"
	"time"
)
import "log"
import "net/rpc"
import "hash/fnv"

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	request := JobRequest{}
	response := JobResponse{}
	for {
		//log.Println("will call master, request", request, "response", response)
		call("Master.GetJob", request, &response)
		log.Println("call master, get", response.Job.FuncName)
		switch response.Job.FuncName {
		case "exit":
			return
		case "wait":
			time.Sleep(time.Second)
		case "map":
			params := MapParams{}
			json.Unmarshal(response.Job.ParamsJson, &params)
			kv := myMapf(mapf, params)
			e := writeFile(kv, fmt.Sprintf("%s-temp", params.FileName))
			if e != nil {
				continue
			}
			request.Job = response.Job
		case "reduce":
			params := ReduceParams{}
			json.Unmarshal(response.Job.ParamsJson, &params)
			kv, e := myReduce(reducef, params)
			if e != nil {
				continue
			}
			e = writeFile(kv, fmt.Sprintf("mr-out-%d", params.MyReduce))
			if e != nil {
				continue
			}
			request.Job = response.Job
		}
	}
}

func myReduce(reducef func(string, []string) string, params ReduceParams) ([]KeyValue, error) {
	okv := []KeyValue{}
	for _, fileName := range params.FileNames {
		kv, e := readFile(fmt.Sprintf("%s-temp", fileName))
		if e != nil {
			log.Fatal(e)
			return nil, e
		}
		okv = append(okv, kv...)
	}
	sort.Sort(ByKey(okv))

	kv := []KeyValue{}
	for i, j := 0, 1; i < len(okv); i, j = j, j+1 {
		for j < len(okv) && okv[j].Key == okv[i].Key {
			j++
		}

		if ihash(okv[i].Key)%params.NReduce == params.MyReduce {
			values := []string{}
			for k := i; k < j; k++ {
				values = append(values, okv[k].Value)
			}

			kv = append(kv, KeyValue{okv[i].Key, reducef(okv[i].Key, values)})
		}
	}
	return kv, nil
}

func myMapf(mapf func(string, string) []KeyValue, params MapParams) []KeyValue {
	kv := mapf(params.FileName, params.Contents)
	sort.Sort(ByKey(kv))
	return kv
	//
	//nkv := []KeyValue{}
	//for i, j := 0, 1; i < len(kv); i, j = j, j+1 {
	//	for j < len(kv) && kv[j].Key == kv[i].Key {
	//		j++
	//	}
	//	value := 0
	//	for k := i; k < j; k++ {
	//		value++
	//	}
	//	nkv = append(nkv, KeyValue{kv[i].Key, strconv.Itoa(value)})
	//}
	//return nkv
}

func writeFile(kvs []KeyValue, fileName string) error {
	tempName := fmt.Sprintf("%s-*", fileName)
	dir, pattern := "", ""
	if i := strings.IndexByte(tempName, '/'); i != -1 {
		//t := strings.SplitAfter(tempName, "\\")
		dir, pattern = tempName[:i+1], tempName[i+1:]
	} else {
		dir, pattern = "", tempName
	}

	tempFile, e := ioutil.TempFile(dir, pattern)
	defer tempFile.Close()
	if e != nil {
		log.Fatal(e)
		return e
	}

	for _, kv := range kvs {
		fmt.Fprintf(tempFile, "%v %v\n", kv.Key, kv.Value)
	}
	os.Rename(tempFile.Name(), fileName)
	return nil
}

func readFile(fileName string) ([]KeyValue, error) {
	file, e := os.Open(fileName)
	defer file.Close()
	if e != nil {
		log.Fatal(e)
		return nil, e
	}

	content, e := ioutil.ReadAll(file)
	if e != nil {
		log.Fatal(e)
		return nil, e
	}

	kv := []KeyValue{}
	for _, line := range strings.Split(string(content), "\n") {
		t := strings.Split(line, " ")
		if len(t) != 2 {
			//log.Println("when readFile, the key,", kv[len(kv)-1].Key, " fileName", fileName)
			break
		}
		kv = append(kv, KeyValue{t[0], t[1]})
	}
	return kv, nil
}

// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
